package spark.requirements

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2

object require1 {
  def main(args: Array[String]): Unit = {
    //1 初始化spark配置信息并建立与spark的连接
    val conf = new SparkConf().setMaster("local[*]").setAppName(require1.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    //2 读取数据
    val RDD1: RDD[String] = sc.textFile("dataSet/user_visit_action.txt")
    //3 将数据封装进用户访问动作样例类
    val RDD2: RDD[UserVisitAction] = RDD1.map(
      line => {
        val fields = line.split("_")
        UserVisitAction(
          fields(0),
          fields(1).toLong,
          fields(2),
          fields(3).toLong,
          fields(4),
          fields(5),
          fields(6).toLong,
          fields(7).toLong,
          fields(8),
          fields(9),
          fields(10),
          fields(11),
          fields(12).toLong
        )
      }
    )
    //需求一
    CategoryTop10App(sc, RDD2)
    //需求二
    SessionTop10App(CategoryTop10App(sc, RDD2), RDD2)
    //需求三
    PageConversionRate(RDD2)
    //关闭sparkcontext连接
    sc.stop()
  }

  //需求一：热门品类Top10
  def CategoryTop10App(sc: SparkContext, userVisitActionRDD: RDD[UserVisitAction]): Array[CategoryCountInfo] = {
    val acc = new CategoryAccumulator
    sc.register(acc)
    userVisitActionRDD.foreach(action => acc.add(action))
    // 1. 把一个品类的三个指标UserVisitAction封装到一个map中
    val cidActionCountGrouped: Map[String, Map[(String, String), Long]] = acc.value.groupBy(_._1._1)
    // 2. 把结果封装到样例类中
    val categoryCountInfoArray: Array[CategoryCountInfo] = cidActionCountGrouped.map {
      case (cid, map) =>
        CategoryCountInfo(cid,
          map.getOrElse((cid, "click"), 0L),
          map.getOrElse((cid, "order"), 0L),
          map.getOrElse((cid, "pay"), 0L)
        )
    }.toArray
    // 3. 对数据进行排序取top10
    val result: Array[CategoryCountInfo] = categoryCountInfoArray
      .sortBy(info => (-info.clickCount, -info.orderCount, -info.payCount))
      .take(10)
    println("需求一：")
    result.foreach(println)

    return result
  }

  //需求二：在需求一的基础上，增加每个品类用户session的点击统计
  //这里我们只关注点击行为
  def SessionTop10App(categoryTop10: Array[CategoryCountInfo], userVisitActionRDD: RDD[UserVisitAction]): Unit = {
    //从需求一获取top10的品类id:List[categoryId]
    val top10Id: List[Long] = categoryTop10.map(_.categoryId.toLong).toList
    //过滤得到热门品类top10的点击行为：RDD(UserVisitAction)
    val filterRDD: RDD[UserVisitAction] = userVisitActionRDD.filter(action => top10Id.contains(action.click_category_id))
    //拼1：RDD[(categoryld, sessionld), 1]
    val addOneRDD: RDD[((Long, String), Int)] = filterRDD.map(action => ((action.click_category_id, action.session_id), 1))
    //累加：RDD[(categoryld,sessionld),count]
    val countRDD: RDD[((Long, String), Int)] = addOneRDD.reduceByKey(_ + _)
    //map:RDD[cid,(sid, count)]
    val mapRDD: RDD[(Long, (String, Int))] = countRDD.map(x => (x._1._1, (x._1._2, x._2)))
    //根据相同品类id将session和对应的点击量分组：RDD[cid, lterable[(sid, count)]]
    val groupRDD: RDD[(Long, Iterable[(String, Int)])] = mapRDD.groupByKey()
    //这种方法是把iterable转成list，然后来排序，但是scala排序需要把list存入内存才能进行排序，如果数据量很大的话内存可能会爆掉
    val result: RDD[(Long, List[(String, Int)])] = groupRDD.mapValues(item => item.toList.sortBy(-_._2).take(10))
    println("需求二：")
    result.foreach(println)
  }

  //需求三：页面单跳转换率统计
  def PageConversionRate(userVisitActionRDD: RDD[UserVisitAction]): Unit = {
    /*算出各个页面的总访问次数*/
    //拼1
    val addOneRDD: RDD[(Long, Long)] = userVisitActionRDD.map(action => (action.page_id, 1L))
    //聚合累加,并先将RDD（弹性分布式数据集）转为Array数组再转为Map
    val pageSum: Map[Long, Long] = addOneRDD.reduceByKey(_ + _).collect().toMap

    /*算出各个页面各个单跳的次数*/
    //先根据session分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = userVisitActionRDD.groupBy(_.session_id)
    //得出同个session的访问页面流程，并转成一次次单跳和拼1
    val pageFlowsRDD: RDD[(String, List[(String, Int)])] = sessionRDD.mapValues(
      datas =>
      {
        //将相同sessionId的数据根据时间升序排序，即是该次session的页面访问顺序
        val userAction: List[UserVisitAction] = datas.toList.sortWith(
          (left, right) => {
            left.action_time < right.action_time
          })
        //只取出页面id即可
        val pageIdsList: List[Long] = userAction.map(_.page_id)
        //拉链操作，将用户访问流程转成一次次单跳 （1->2->3->4 ----> 1->2,2->3,3->4）
        //这里的.tail就是取这个list除了第一个元素的其他元素
        //[1,2,3,4].zip[2,3,4]=1->2,2->3,3->4
        val pageFlows: List[(Long, Long)] = pageIdsList.zip(pageIdsList.tail)
        val pageFlowsAddOne: List[(String, Int)] = pageFlows.map {
          case (pageId1, pageId2) => (pageId1 + "->" + pageId2, 1)
        }
        pageFlowsAddOne
      })
    //提取出单跳流和对应的次数，不需要sessionid了
    val pageFlowsMapRDD: RDD[(String, Int)] = pageFlowsRDD.map(_._2).flatMap(list => list)
    //聚合累加求出各个单跳流的总次数
    val pageFlowsSumRDD: RDD[(String, Int)] = pageFlowsMapRDD.reduceByKey(_ + _)

    /*页面单跳转换率的计算 （各个页面各个单跳的次数/各个页面的总访问次数)  */
    println("需求三：")
    pageFlowsSumRDD.foreach {
      case (pageFlow, fz) => {
        val pageIds: Array[String] = pageFlow.split("->")
        val fmPageId: Long = pageIds(0).toLong
        val fmSum: Long = pageSum.getOrElse(fmPageId, 1L)
        //转换率结果
        println(pageFlow + "的页面单跳转换率为:" + fz.toDouble / fmSum)
      }
    }

  }

  //
}

//用户访问动作样例类
case class UserVisitAction(
                            date: String, //用户点击行为的日期
                            user_id: Long, //用户的ID
                            session_id: String, //Session的ID
                            page_id: Long, //某个页面的ID
                            action_time: String, //动作的时间点
                            search_keyword: String, //用户搜索的关键词
                            click_category_id: Long, //某一个商品品类的ID
                            click_product_id: Long, //某一个商品的ID
                            order_category_ids: String, //一次订单中所有品类的ID集合
                            order_product_ids: String, //一次订单中所有商品的ID集合
                            pay_category_ids: String, //一次支付中所有品类的ID集合
                            pay_product_ids: String, //一次支付中所有商品的ID集合
                            city_id: Long //城市 id
                          )

//累加结果样例类
case class CategoryCountInfo(
                              categoryId: String, //品类id
                              clickCount: Long, //点击数
                              orderCount: Long, //下单数
                              payCount: Long //支付数
                            )

//自定义累加器(需求一)
class CategoryAccumulator extends AccumulatorV2[UserVisitAction, Map[(String, String), Long]] {
  //输入是用户访问动作样例类，输出是Map[（品类id,点击/下单/支付/），次数]
  private var map = Map[(String, String), Long]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[UserVisitAction, Map[(String, String), Long]] = {
    val acc = new CategoryAccumulator
    acc.map = map
    acc
  }

  override def reset(): Unit = Map[(String, String), Long]()

  override def add(v: UserVisitAction): Unit = {
    v match {
      //点击
      case action if action.click_category_id != -1 => {
        val key: (String, String) = (action.click_category_id.toString, "click")
        map += key -> (map.getOrElse(key, 0L) + 1L)
      }
      //下单
      case action if action.order_category_ids != "null" => {
        val cids: Array[String] = action.order_category_ids.split(",")
        cids.foreach(
          cid => {
            val key: (String, String) = (cid, "order")
            map += key -> (map.getOrElse(key, 0L) + 1L)
          }
        )
      }
      //支付
      case action if action.pay_category_ids != "null" => {
        val pids: Array[String] = action.pay_category_ids.split(",")
        pids.foreach(pid => {
          val key: (String, String) = (pid, "pay")
          map += key -> (map.getOrElse(key, 0L) + 1L)
        }
        )
      }
      case _ =>
    }
  }

  override def merge(other: AccumulatorV2[UserVisitAction, Map[(String, String), Long]]): Unit = {
    map = other match {
      case o: CategoryAccumulator =>
        o.map.foldLeft(map) {
          // case出来的任何东西都不能改, 只能读
          case (map, (key, count)) => //(
            // 当前map，o.map)--->(当前map，(key,value))--->(当前map，((品类ID，对应的行为),对应行为的count))
            //直接返回新的集合就可以了
            map + (key -> (map.getOrElse(key, 0L) + count))
        }
      case _ =>
        throw new UnsupportedOperationException
    }
  }

  override def value: Map[(String, String), Long] = map

}

